-module(ham_topic).
-author("hpj").

-import(proplists, [get_value/2, get_value/3]).

-include_lib("emysql/include/emysql.hrl").

-include("mysql.hrl").

-include("emqttd.hrl").

-include("emqttd_topic.hrl").

-include("emqttd_packet.hrl").

%% API
-export([handle/1]).

%% topic inner
handle(Params) ->
  ActionType = get_value(<<"action_type">>,Params),
  handle(ActionType,Params).

%%topic subscriber or unsubscriber
handle(1,Data) ->
  SendUid = get_value(<<"send_uid">>,Data),
  Subject = get_value(<<"subject">>,Data),
  NewSubject = <<<<"p@">>/binary,Subject/binary>>,
  QoS = get_value(<<"qos">>,Data),
  Type = get_value(<<"type">>,Data),
  Fun = 
    case Type of
      0 -> 
        fun() ->
          ID = {Subject,SendUid},
          case mnesia:dirty_match_object(topic_subscriber,#topic_subscriber{topic = NewSubject,_ = '_'}) of 
            [] ->
              mnesia:delete(topic,NewSubject,write);
            [_Head|Tail] ->
              case Tail of 
                [] ->
                  mnesia:delete(topic_subscriber,ID,write),
                  mnesia:delete(topic,NewSubject,write);
                _ ->
                   mnesia:delete(topic_subscriber,ID,write)
              end    
          end
        end;
      1 ->
        fun() ->
          TopicRecord = #topic{topic = NewSubject},
          mnesia:write(topic, TopicRecord, write), 
          Topic_subscriberID = {NewSubject,SendUid},
          SubscriberRecord = #topic_subscriber{id =Topic_subscriberID,topic = NewSubject,qos = QoS,user_id = SendUid},
          mnesia:write(topic_subscriber, SubscriberRecord, write)
        end
    end,    
  case mnesia:activity(sync_transaction, Fun, [], mnesia_frag) of
      {'exist',_} ->
        {error, [{<<"code">>,5}]};
      _ ->
        {ok, [{<<"code">>,0}]}
  end;

%% topic publish
handle(2, Data) ->
  {ok,Im} = application:get_env(ham_server,emqttd),
  Send = get_value(<<"send_uid">>,Data),
  Content = get_value(<<"content">>,Data),
  Subject = get_value(<<"subject">>,Data),
  NewSubject = <<<<"p@">>/binary,Subject/binary>>,
  PacketId = 
    case gen_server:call(package_ets,{find,Send}) of 
      [] -> 
        gen_server:cast(package_ets,{insert,Send,1}),
        1;
      [Value|_] when Value >= 65535 ->
        Value1 = 1,
        gen_server:cast(package_ets,{delete,Send}),
        gen_server:cast(package_ets,{insert,Send,Value1}),
        Value1;
      [Value|_] ->
        Value1 = Value + 1,
        gen_server:cast(package_ets,{delete,Send}),
        gen_server:cast(package_ets,{insert,Send,Value1}),
        Value1
    end,
  rpc:cast(Im, emqttd_ham, to_group_ham, [Send, NewSubject, PacketId, base64:decode(Content)]),       
  {ok, [{<<"code">>,0}]};



handle(_Method, Req) ->
  Response = jsx:encode([{<<"ret">>,4},{<<"message">>,<<"api invalid">>}]),
  Req:ok({"text/plan",Response}).